'''
1、接受tessng向kafka发送的车辆数据,并进行展示


'''
import threading
from concurrent.futures import ThreadPoolExecutor

import carla
import numpy as np
import time

import redis
from carla import Transform, Rotation, Location
from kafka import KafkaConsumer, KafkaProducer
import json
from pyproj import Proj


class myCarla:
    def __init__(self, carla_ip, carla_port):
        # 创建carla客户端

        self.client = carla.Client(carla_ip, carla_port)
        self.client.set_timeout(10.0)
        self.world = self.client.load_world('tongji')
        # carla中运行的车，key-已经生成的车辆id，value-carla中id
        self.run_carla_id_dict = {}
        # 车辆类型和carla类型字典关系，key-车辆类型，value-carla类型
        self.now_car_type_carla_type_dist = {
            1: 'vehicle.lincoln.mkz_2017',
            2: 'vehicle.bmw.grandtourer'
        }
        # 默认车辆类型
        self.default_carla_type = 'vehicle.lincoln.mkz_2017'
        # 转换坐标
        self.rule = Proj("+proj=tmerc +lon_0=121.2092870660126 +lat_0=31.292829882838856 +ellps=WGS84")
        # 获取carla中的交通管理者
        self.traffic_manager = self.client.get_trafficmanager(8000)
        # ==================kafka======================
        # self.bootstrap_servers = '106.120.201.126:14576'
        # # 初始化 Kafka 生产者
        # self.kafka_producer = KafkaProducer(
        #     bootstrap_servers=self.bootstrap_servers,
        #     value_serializer=lambda v: json.dumps(v).encode('utf-8')  # 将数据序列化为 JSON
        # )
        # ==================redis======================
        self.redis_host = '10.100.12.58'
        self.redis_port = 6378
        self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port, db=0)

    # 设置天气
    def set_weather(self):
        weather = carla.WeatherParameters(
            cloudiness=0.0,  # 0-100  0 是晴朗的天空，100 是完全阴天
            precipitation=0.0,  # 0 表示没有下雨，100 表示大雨
            precipitation_deposits=0.0,  # 0 表示道路上没有水坑，100 表示道路完全被雨水覆盖
            wind_intensity=0.0,  # 0 表示平静，100 表示强风，风会影响雨向和树叶
            sun_azimuth_angle=0.0,  # 太阳方位角，0～360
            sun_altitude_angle=90.0,  # 太阳高度角，90 是中午，-90 是午夜
            fog_density=0.0,  # 0～100 表示雾的浓度或厚度，仅影响RGB相机传感器
            fog_distance=0.0,  # 雾开始的距离，单位为米
            wetness=0.0,  # 0～100 表示道路湿度百分比，仅影响RGB相机传感器
            fog_falloff=0.0,  # 雾的密度，0至无穷大，0 表示雾比空气轻，覆盖整个场景，1表示与空气一样，覆盖正常大小的建筑物
            scattering_intensity=0.0,  # 控制光线对雾的穿透程度
            mie_scattering_scale=0.0,  # 控制光线与花粉或空气等大颗粒的相互作用，导致天气朦胧，光源周围有光晕，0表示无影响
            rayleigh_scattering_scale=0.0331,  # 控制光与空气分子等小粒子的相互作用，取决于光波长，导致白天蓝天或晚上红天
        )
        self.world.set_weather(weather)

    # 设置视角坐标
    def set_view(self, lon, lat):
        x, y = self.rule(lon, lat)
        z = 140
        view_angle = Transform(Location(x=x, y=-y, z=z),
                               Rotation(pitch=-45, yaw=-180, roll=0.000000))
        self.world.get_spectator().set_transform(view_angle)  # 设置世界视角

    # 创建交通管理器对象
    def set_traffic_manager(self):
        self.traffic_manager.set_global_distance_to_leading_vehicle(2.0)
        self.traffic_manager.global_percentage_speed_difference(40.0)
        settings = self.world.get_settings()
        self.traffic_manager.set_synchronous_mode(True)
        if not settings.synchronous_mode:
            print("开启异步模式")
            synchronous_master = True
            settings.synchronous_mode = False
            settings.fixed_delta_seconds = 0.1
            self.world.apply_settings(settings)

    # 获取carla车辆
    def get_carla_car_type(self, veh_type):
        if veh_type in self.now_car_type_carla_type_dist.keys():
            carla_type = self.now_car_type_carla_type_dist[veh_type]
        else:
            carla_type = self.default_carla_type
        carla_car = self.world.get_blueprint_library().find(carla_type)
        carla_car.set_attribute('color', '0,0,0')
        return carla_car

    # 获取carla车辆信息
    @staticmethod
    def get_vehicle_info(vehicle):
        transform = vehicle.get_transform()
        velocity = vehicle.get_velocity()
        return {
            'id': vehicle.id,
            'type_id': vehicle.type_id,
            'location': {
                'x': transform.location.x,
                'y': transform.location.y,
                'z': transform.location.z
            },
            'rotation': {
                'pitch': transform.rotation.pitch,
                'yaw': transform.rotation.yaw,
                'roll': transform.rotation.roll
            },
            'velocity': {
                'x': velocity.x,
                'y': velocity.y,
                'z': velocity.z
            }
        }

    # 通过kafka，获取车辆信息并设置
    def get_set_car_for_kafka(self):
        consumer = KafkaConsumer(
            'targetChannel',
            bootstrap_servers=self.bootstrap_servers,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        while True:
            for message in consumer:
                if message:
                    value = message.value.decode('utf-8')
                    json_data = json.loads(value)
                    participants = json_data['value']['value']
                    # 当前帧创建失败的车辆id集合
                    now_error_id_list = []
                    # 当前帧所有车辆id集合
                    now_id_list = []
                    for p in participants:
                        now_id_list.append(p["id"])
                    # 添加交通流
                    for par in participants:
                        x, y = self.rule(par["longitude"], par["latitude"])
                        veh_type = par['vehicleType']
                        car_id = par['id']
                        angle = par["courseAngle"]
                        # 判断是否已经生成过该车辆
                        if car_id not in self.run_carla_id_dict.keys():
                            # 没有生成过，则生成
                            trans = Transform(carla.Location(x=x, y=-y, z=15), carla.Rotation(yaw=angle - 90))
                            # 根据车辆类型生成车辆
                            carla_car = self.get_carla_car_type(veh_type)
                            # 将车辆生成请求添加到批量请求中
                            batch = [
                                carla.command.SpawnActor(carla_car, trans).then(
                                    carla.command.SetSimulatePhysics(carla.command.FutureActor, False))
                            ]
                            # 将车辆生成请求发送到服务器端，并获取响应
                            response = self.client.apply_batch_sync(batch, False)[0]
                            if response.error:
                                now_error_id_list.append(car_id)
                            else:
                                self.run_carla_id_dict[car_id] = response.actor_id
                        else:
                            # 生成过，则更新位置和角度
                            # 更新位置和角度
                            trans = Transform(carla.Location(x=x, y=-y, z=15),
                                              carla.Rotation(yaw=par["courseAngle"] - 90))
                            carla_id = self.run_carla_id_dict[car_id]
                            carla_car = self.world.get_actor(carla_id)
                            if carla_car is not None:
                                carla_car.set_transform(trans)
                    print(f'当前carla中运行车辆个数：{len(self.run_carla_id_dict)}')
                    # 销毁不在当前帧的id
                    destroy_id_list = [id for id in self.run_carla_id_dict if id not in now_id_list]
                    for destroy_id in destroy_id_list:
                        carla_id = self.run_carla_id_dict[destroy_id]
                        vehicle = self.world.get_actor(carla_id)
                        if vehicle is not None:
                            vehicle.destroy()
                        del self.run_carla_id_dict[destroy_id]
                    print(f'本次更新数量：{len(now_id_list)}，销毁数量：{len(destroy_id_list)}')

    # 发送数据到kafka
    def send_data_for_kafka(self, send_topic):
        while True:
            # 获取场景中的所有车辆
            carla_veh_list = self.world.get_actors().filter('vehicle.*')
            vehicle_list = []
            # 遍历所有车辆并获取信息
            for vehicle in carla_veh_list:
                vehicle_info = self.get_vehicle_info(vehicle)
                vehicle_list.append(vehicle_info)
            # print(vehicle_list)
            self.kafka_producer.send(topic=send_topic, value=vehicle_list, key='key')
            # 休眠一段时间，避免频繁发送
            time.sleep(1)

    # 通过redis，获取车辆信息并设置
    def get_set_car_for_redis(self, redis_channel):
        pubsub = self.redis_client.pubsub()
        pubsub.subscribe(redis_channel)
        for message in pubsub.listen():
            if message['type'] == 'message':
                # 接收到数据，添加处理逻辑
                receive_data = message['data'].decode('utf-8')
                data = json.loads(receive_data)
                participants = data['value']['value']
                # 当前帧创建失败的车辆id集合
                now_error_id_list = []
                # 当前帧所有车辆id集合
                now_id_list = []
                for p in participants:
                    now_id_list.append(p["id"])
                # 添加交通流
                for par in participants:
                    x, y = self.rule(par["longitude"], par["latitude"])
                    veh_type = par['vehicleType']
                    car_id = par['id']
                    angle = par["courseAngle"]
                    # 判断是否已经生成过该车辆
                    if car_id not in self.run_carla_id_dict.keys():
                        # 没有生成过，则生成
                        trans = Transform(carla.Location(x=x, y=-y, z=15), carla.Rotation(yaw=angle - 90))
                        # 根据车辆类型生成车辆
                        carla_car = self.get_carla_car_type(veh_type)
                        # 将车辆生成请求添加到批量请求中
                        batch = [
                            carla.command.SpawnActor(carla_car, trans).then(
                                carla.command.SetSimulatePhysics(carla.command.FutureActor, False))
                        ]
                        # 将车辆生成请求发送到服务器端，并获取响应
                        response = self.client.apply_batch_sync(batch, False)[0]
                        if response.error:
                            now_error_id_list.append(car_id)
                        else:
                            self.run_carla_id_dict[car_id] = response.actor_id
                    else:
                        # 生成过，则更新位置和角度
                        # 更新位置和角度
                        trans = Transform(carla.Location(x=x, y=-y, z=15),
                                          carla.Rotation(yaw=par["courseAngle"] - 90))
                        carla_id = self.run_carla_id_dict[car_id]
                        carla_car = self.world.get_actor(carla_id)
                        if carla_car is not None:
                            carla_car.set_transform(trans)
                print(f'当前carla中运行车辆个数：{len(self.run_carla_id_dict)}')
                # 销毁不在当前帧的id
                destroy_id_list = [id for id in self.run_carla_id_dict if id not in now_id_list]
                for destroy_id in destroy_id_list:
                    carla_id = self.run_carla_id_dict[destroy_id]
                    vehicle = self.world.get_actor(carla_id)
                    if vehicle is not None:
                        vehicle.destroy()
                    del self.run_carla_id_dict[destroy_id]
                print(f'本次更新数量：{len(now_id_list)}，销毁数量：{len(destroy_id_list)}')

    # 发送数据到redis
    def send_data_for_redis(self, send_channel):
        while True:
            # 获取场景中的所有车辆
            carla_veh_list = self.world.get_actors().filter('vehicle.*')
            vehicle_list = []
            # 遍历所有车辆并获取信息
            for vehicle in carla_veh_list:
                vehicle_info = self.get_vehicle_info(vehicle)
                vehicle_list.append(vehicle_info)
            self.redis_client.publish(send_channel, json.dumps(vehicle_list))
            # 休眠一段时间，避免频繁发送
            time.sleep(0.1)


if __name__ == "__main__":
    executor = ThreadPoolExecutor(max_workers=3)
    myCarla = myCarla('localhost', 2000)
    myCarla.set_weather()
    myCarla.set_view(121.20322248292439, 31.292553908303493)
    myCarla.set_traffic_manager()
    # 接收并设置背景车辆信息
    redis_get_channel = 'bjc_channel'
    executor.submit(myCarla.get_set_car_for_redis, redis_get_channel)
    # 接收并设置主车辆信息
    redis_get_channel = 'target_channel'
    executor.submit(myCarla.get_set_car_for_redis, redis_get_channel)
    # carla中车辆信息发送数据到kafka
    redis_send_channel = 'carla_info'
    executor.submit(myCarla.send_data_for_redis, redis_send_channel)
